package com.test.reactor.utils;

import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;

import com.test.reactor.model.mongo.Message;
import com.test.reactor.repository.mongo.MessageRepository;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.socket.WebSocketSession;

import com.test.reactor.actor.UserActor;
import com.test.reactor.ws.MyMessage;

import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.actor.ActorSystem;
import akka.actor.Props;
import reactor.core.publisher.Mono;
import scala.concurrent.duration.FiniteDuration;

@Service
@Slf4j
public class ActorUtils {
    public static String localAddress ;

	@Autowired
    private ActorSystem actorSystem;
    @Autowired
	private ActorRef receiveActorRef;
    @Autowired
	private MessageRepository messageRepository;
	@Autowired
	RedisTemplate redisTemplate;
    
	public void stopActor(String userName) {
		ActorSelection actorSelection = actorSystem.actorSelection("akka://system/user/" + "us-" + userName);
		CompletionStage<ActorRef> resolveOneCS = actorSelection.resolveOneCS(new FiniteDuration(3, TimeUnit.MINUTES));
		Mono.fromCompletionStage(resolveOneCS).doOnError(t -> {
			System.out.println(userName + "---不在线");
		}).subscribe(act -> {
			actorSystem.stop(act);
		});
	}
	public void sendMessage(MyMessage myMessage) {
		saveMessageToMongo(myMessage);

	    String userName = "us-" + myMessage.getTo();
        String address = redisTemplate.opsForHash().get("user","userAddress:" + userName)+"";
        log.info(userName + "  address: " + address);
        ActorSelection actorSelection = null;
        if (StringUtils.equals(address, localAddress))
            actorSelection = actorSystem.actorSelection("akka://system/user/" + userName);
        else {
            actorSelection = actorSystem.actorSelection("akka.tcp://system@"+address+"/user/" + userName);
        }
		CompletionStage<ActorRef> resolveOneCS = actorSelection.resolveOneCS(new FiniteDuration(3, TimeUnit.MINUTES));
		Mono.fromCompletionStage(resolveOneCS).doOnError(t -> {
			System.out.println(myMessage.getTo() + "---不在线--"+ myMessage);
		}).subscribe(act -> {
			act.tell(myMessage, receiveActorRef);
		});
	}
	public void creatActor(String user, WebSocketSession session2) {
		actorSystem.actorOf(Props.create(UserActor.class, session2), "us-" + user);
		redisTemplate.opsForHash().put("user","userAddress:us-" + user, localAddress);
	}
	public void removeRedisUser(String name){
		redisTemplate.opsForHash().delete("user","userAddress:us-" + name);
	}

	private void saveMessageToMongo(MyMessage myMessage){
		Message msg = new Message();
		msg.setFrom(myMessage.getFrom());
		msg.setMsg(myMessage.getMsg());
		msg.setTo(myMessage.getTo());
		msg.setMsgId(myMessage.getMsgId());
		Mono<Message> save = messageRepository.save(msg);
		save.subscribe(message -> log.info("message:" + message.getMsgId()));
	}
}
